package dyyx.delay;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

public class Demo {

	public static void main(String[] args) throws Exception {
		String hostname = "localhost";
		int port = 9001;

		System.out.println("hostname=" + hostname + ",port=" + port);

		// 1 获取运行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		// 2 连接socket获取输入的数据
		DataStream<String> inputStream = env.socketTextStream(hostname, port);

		OutputTag<SensorReading> outputTag = new OutputTag<SensorReading>("side");

		DataStream<SensorReading> dataStream = inputStream.map(data -> {
			String[] dataArray = data.split(",");
			SensorReading sensorReading = new SensorReading();
			sensorReading.id = dataArray[0];
			sensorReading.timestamp = Long.parseLong(dataArray[1]);
			sensorReading.temperature = Double.parseDouble(dataArray[2]);
			return sensorReading;

		}).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)) {
			public long extractTimestamp(SensorReading element) {
				return element.timestamp * 1000;
			}
		});

		DataStream<SensorReading> minStream = dataStream.keyBy("id")
				// .window( SlidingEventTimeWindows.of(Time.seconds(10),Time.seconds(2)))
				.timeWindow(Time.seconds(10)).allowedLateness(Time.seconds(4)).sideOutputLateData(outputTag)
				.minBy("temperature");

		dataStream.print("data");

		minStream.print("min");

		// minStream.getSideOutput(outputTag).print("slide");

		env.execute("demo1");

	}
}
